package org.example.realtime.gongdan.log.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.PathNotFoundException;
import org.apache.flink.table.functions.ScalarFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义UDF函数，模拟MySQL的get_json_string功能
 * 使用示例：get_json_string(json_str, '$.field.subfield')
 */
public class JsonExtractorUDF extends ScalarFunction {
    private static final Logger LOG = LoggerFactory.getLogger(JsonExtractorUDF.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 从JSON字符串中提取指定路径的值
     * @param jsonStr JSON字符串
     * @param pathExpr JSON路径表达式
     * @return 提取的字符串值，如果路径不存在返回null
     */
    public String eval(String jsonStr, String pathExpr) {
        if (jsonStr == null || pathExpr == null) {
            return null;
        }

        try {
            // 使用JsonPath提取值
            Object result = JsonPath.read(jsonStr, pathExpr);

            // 处理不同类型的返回值
            if (result == null) {
                return null;
            } else if (result instanceof String) {
                return (String) result;
            } else {
                // 对于非字符串类型，转换为JSON字符串
                return objectMapper.writeValueAsString(result);
            }
        } catch (PathNotFoundException e) {
            LOG.debug("Path not found in JSON: {}", pathExpr);
            return null;
        } catch (Exception e) {
            LOG.warn("Error parsing JSON or extracting path: {}", pathExpr, e);
            return null;
        }
    }

    // 可选：为Flink提供函数签名信息
    @Override
    public String toString() {
        return "get_json_string(json_str, path_expr)";
    }
}
